package article

import (
	"context"
	"forum/internal/common"
	"forum/internal/domain/event/article"
	"forum/intr/domain/repo"
	"forum/pkg/saramax"
	"github.com/IBM/sarama"
	"github.com/prometheus/client_golang/prometheus"
	"time"
)

type BatchReadCntConsumer struct {
	client    sarama.Client
	interRepo repo.InteractiveRepository
	opt       prometheus.SummaryOpts
}

func NewBatchReadCntConsumer(client sarama.Client, interRepo repo.InteractiveRepository) *BatchReadCntConsumer {
	return &BatchReadCntConsumer{client: client, interRepo: interRepo}
}

func (b *BatchReadCntConsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("intr", b.client)
	if err != nil {
		return err
	}
	go func() {
		er := cg.Consume(context.Background(),
			[]string{common.TopicReadEvent},
			saramax.NewBatchHandler[article.ReadEvent](b.Consume))
		if er != nil {
			//TODO log
		}
	}()
	return err
}

func (b *BatchReadCntConsumer) Consume(msgs []*sarama.ConsumerMessage, events []article.ReadEvent) error {
	bizs := make([]string, 0, len(events))
	bizIds := make([]int64, 0, len(events))
	for _, evt := range events {
		bizs = append(bizs, "article")
		bizIds = append(bizIds, evt.Aid)
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	return b.interRepo.BatchReadCnt(ctx, bizs, bizIds)
}
